package com.qianbao.finance.sync.modules.elasticsearch.websocket.server;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import com.qianbao.finance.sync.common.utils.SpringContextUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.handler.codec.http.websocketx.TextWebSocketFrame;

/**
 * 聊天室
 * 
 * @author WangYuan
 * @date 2016年7月12日下午3:10:38
 */
public class ChatRoom {

    public static final String DEFAULT_GROUP = "default_group";

    private final static Logger logger = Logger.getLogger(ChatRoom.class);

    private static ShareConfig shareConfig = (ShareConfig) SpringContextUtils.getBean("shareConfig");
    /**
     * 聊天室名字
     */
    private String chatRoomName;

    /**
     * 子聊天室和用户连接映射表
     */
    private Map<String, Set<ChannelHandlerContext>> groupsChannelHandlerContext =
            new ConcurrentHashMap<String, Set<ChannelHandlerContext>>();
    /**
     * 连接最后心跳时间：客户端定时发送join表明心跳，长期未检测到心跳，则被清理
     */
    private Map<Integer, Long> channelHandlerContextLastHeartBeat =
            new ConcurrentHashMap<Integer, Long>();

    /**
     * 清理过期会话对象
     * 
     */
    public void cleanTimeoutSession() {
        List<ChannelHandlerContext> clearChannelHandlerContexts =
                new ArrayList<ChannelHandlerContext>();
        try {
            final long now = System.currentTimeMillis();
            Set<String> ks = groupsChannelHandlerContext.keySet();
            List<String> groupsChannelHandlerContextList = new ArrayList<String>();
            groupsChannelHandlerContextList.addAll(ks);
            for (String group : groupsChannelHandlerContextList) {
                Set<ChannelHandlerContext> channelHandlerContexts =
                        groupsChannelHandlerContext.get(group);
                // 防止迭代的时候修改
                List<ChannelHandlerContext> channelHandlerContextList =
                        new ArrayList<ChannelHandlerContext>();
                synchronized (channelHandlerContexts) {
                    channelHandlerContextList.addAll(channelHandlerContexts);
                }
                if (channelHandlerContexts != null) {
                    for (ChannelHandlerContext channelHandlerContext : channelHandlerContextList) {
                        try {
                            if (channelHandlerContextLastHeartBeat.get(channelHandlerContext
                                    .hashCode()) == null
                                    || now
                                            - channelHandlerContextLastHeartBeat
                                                    .get(channelHandlerContext.hashCode()) > shareConfig.getConnClearPeriod()
                                    || !channelHandlerContext.getChannel().isOpen()
                                    || !channelHandlerContext.getChannel().isConnected()) {
                                StringBuffer sf = new StringBuffer();
                                sf.append(chatRoomName).append(
                                        "心跳超时，清理:  " + channelHandlerContext.hashCode());
                                logger.info(sf.toString());
                                clearChannelHandlerContexts.add(channelHandlerContext);
                            }
                        } catch (Exception ee) {
                            try {
                                logger.error(
                                        channelHandlerContext.hashCode()
                                                + ""
                                                + channelHandlerContextLastHeartBeat
                                                        .get(channelHandlerContext.hashCode()) + ""
                                                + channelHandlerContext.getChannel() + "", ee);
                            } catch (Exception e) {
                                logger.error("", e);
                            }
                            clearChannelHandlerContexts.add(channelHandlerContext);
                        }
                    }
                }
            }
            if (logger.isInfoEnabled()) {
                StringBuffer sf = new StringBuffer();
                sf.append(this.chatRoomName).append(
                        "需要清理消息数目:  " + clearChannelHandlerContexts.size());
                logger.info(sf.toString());
            }
            this.clearIt(clearChannelHandlerContexts);
        } catch (Exception e) {
            logger.error("", e);
        }
    }

    private void clearIt(List<ChannelHandlerContext> clearChannelHandlerContexts) {
        if (clearChannelHandlerContexts.size() > 0) {
            for (ChannelHandlerContext channelHandlerContext : clearChannelHandlerContexts) {
                try {
                    if (logger.isInfoEnabled()) {
                        StringBuffer sf = new StringBuffer();
                        sf.append(this.chatRoomName).append(
                                "清理: " + channelHandlerContext.hashCode());
                        logger.info(sf.toString());
                    }
                    ShareBean.getInstance().closeChannelHandlerContext(channelHandlerContext);
                } catch (Exception e) {
                    logger.error("", e);
                }
            }
            clearChannelHandlerContexts.clear();
        }
    }

    /**
     * 加入聊天室
     * 
     * @param channelHandlerContext
     */
    public void addChannelHandlerContext(String groups, ChannelHandlerContext channelHandlerContext) {
        channelHandlerContextLastHeartBeat.put(channelHandlerContext.hashCode(),
                System.currentTimeMillis());
        if (logger.isDebugEnabled()) {
            logger.debug(chatRoomName + " " + groups + "有会话加入聊天室: "
                    + channelHandlerContext.hashCode());
        }
        // 没有设置分组则列入全局分组
        if (groups == null || groups.equals("")) {
            groups = DEFAULT_GROUP;
        }

        String[] groupArray = StringUtils.split(groups, ",");
        for (String group : groupArray) {
            Set<ChannelHandlerContext> set = null;
            set = groupsChannelHandlerContext.get(group);
            if (set == null) {
                // 分组不存在，创建
                logger.info("子分组: " + group + "不存在，需要创建");
                set = new HashSet<ChannelHandlerContext>();
                logger.debug("groupsChannelHandlerContext size: "
                        + groupsChannelHandlerContext.size());
                groupsChannelHandlerContext.put(group, set);
                logger.debug("after put,groupsChannelHandlerContext size: "
                        + groupsChannelHandlerContext.size());
            }
            synchronized (set) {
                if (set.contains(channelHandlerContext) == false) {
                    // 重新加入
                    set.add(channelHandlerContext);
                    if (logger.isInfoEnabled()) {
                        logger.info(chatRoomName + " " + groups + "成功加入聊天室: "
                                + channelHandlerContext.hashCode());
                    }
                }
            }
        }

    }

    public void removeChannelHandlerContext(ChannelHandlerContext channelHandlerContext) {
        List<String> clearGroupList = new ArrayList<String>();
        logger.info(chatRoomName + " 有会话被移出聊天室: " + channelHandlerContext.hashCode());
        Set<String> ks = groupsChannelHandlerContext.keySet();
        for (String key : ks) {
            // 查找每一个分组
            Set<ChannelHandlerContext> set = groupsChannelHandlerContext.get(key);
            synchronized (set) {
                if (set.contains(channelHandlerContext)) {
                    set.remove(channelHandlerContext);
                    if (set.isEmpty()) {
                        // 当分组没有任何用户时，列入为清理
                        clearGroupList.add(key);
                    }
                }
            }
        }
        for (String key : clearGroupList) {
            // 清理空的分组
            if (logger.isInfoEnabled()) {
                logger.info("清理：" + key);
            }
            logger.debug("groupsChannelHandlerContext size: " + groupsChannelHandlerContext.size());
            groupsChannelHandlerContext.remove(key);
            logger.debug("after remove,groupsChannelHandlerContext size: "
                    + groupsChannelHandlerContext.size());
        }
        clearGroupList.clear();
        channelHandlerContextLastHeartBeat.remove(channelHandlerContext.hashCode());
    }

    /**
     * 给聊天室发送消息
     * 
     * @param msg
     * @param groups 子房间，当子房间为用户编号时，表明对用户广播
     * @throws Exception
     */
    public void broadcastMessage(String groups, final String msg) throws Exception {
        final List<ChannelHandlerContext> _clearChannelHandlerContexts =
                new ArrayList<ChannelHandlerContext>();
        if (logger.isInfoEnabled()) {
            logger.info("groups: " + groups);
            logger.info("msg: " + msg);
        }

        // 不指定分组时，列入到全局分组
        if (groups == null || groups.equals("")) {
            groups = DEFAULT_GROUP;
        }
        final TextWebSocketFrame textWebSocketFrame = new TextWebSocketFrame(msg);
        String[] groupArray = StringUtils.split(groups, ",");
        // 发送过此消息的用户列表，用于判断是否重复发送
        final Set<Integer> sendChannelHandlerContext = new HashSet<Integer>();
        logger.debug("broadcastMessage groupsChannelHandlerContext size: "
                + groupsChannelHandlerContext.size());
        for (String group : groupArray) {
            Set<ChannelHandlerContext> channelHandlerContexts =
                    groupsChannelHandlerContext.get(group);
            if (null == channelHandlerContexts) {
                logger.debug("broadcastMessage channelHandlerContexts is null");
                throw new Exception("broadcastMessage channelHandlerContexts is null");
            }
            List<ChannelHandlerContext> channelHandlerContextList =
                    new ArrayList<ChannelHandlerContext>();
            // synchronized (channelHandlerContexts) {
            channelHandlerContextList.addAll(channelHandlerContexts);
            // }
            for (ChannelHandlerContext channelHandlerContext : channelHandlerContextList) {
                int hashCode = channelHandlerContext.hashCode();
                if (sendChannelHandlerContext.contains(hashCode) == true) {
                    // 已经发送过的不再重复发送，防止同一个用户订阅两个分组，从而收到2个相同的消息。
                    continue;
                }
                sendChannelHandlerContext.add(hashCode);
                // ShareBean.getInstance().getChatRoomExecutorService()
                // .execute(new Runnable() {
                // @Override
                // public void run() {
                try {
                    Channel channel = channelHandlerContext.getChannel();
                    if (channel.isOpen() && channel.isConnected()) {
                        if (logger.isDebugEnabled()) {
                            logger.debug("发送消息：" + channelHandlerContext.hashCode());
                        }
                        channel.write(textWebSocketFrame);
                    }
                } catch (Exception e) {
                    logger.error("发送消息异常" + channelHandlerContext.hashCode(), e);
                    _clearChannelHandlerContexts.add(channelHandlerContext);
                }
                // }
                // });
            }
        }
        if (logger.isInfoEnabled()) {
            StringBuffer sf = new StringBuffer();
            sf.append(this.chatRoomName).append("发送消息完毕 ");
            logger.info(sf.toString());
        }
        this.clearIt(_clearChannelHandlerContexts);
    }

    /*
     * public void broadcastMessage(String groups, final String msg) { final
     * List<ChannelHandlerContext> _clearChannelHandlerContexts = new
     * ArrayList<ChannelHandlerContext>(); if (logger.isInfoEnabled()) { logger.info("groups: " +
     * groups); logger.info("msg: " + msg); }
     * 
     * // 不指定分组时，列入到全局分组 if (groups == null || groups.equals("")) { groups = DEFAULT_GROUP; } final
     * TextWebSocketFrame textWebSocketFrame = new TextWebSocketFrame(msg); String[] groupArray =
     * StringUtils.split(groups, ","); // 发送过此消息的用户列表，用于判断是否重复发送 final Set<Integer>
     * sendChannelHandlerContext = new HashSet<Integer>(); for (String group : groupArray) {
     * Set<ChannelHandlerContext> channelHandlerContexts = groupsChannelHandlerContext.get(group);
     * List<ChannelHandlerContext> channelHandlerContextList = new
     * ArrayList<ChannelHandlerContext>(); synchronized (channelHandlerContexts) {
     * channelHandlerContextList.addAll(channelHandlerContexts); } for (ChannelHandlerContext
     * channelHandlerContext : channelHandlerContextList) { int hashCode =
     * channelHandlerContext.hashCode(); if (sendChannelHandlerContext.contains(hashCode) == true) {
     * // 已经发送过的不再重复发送，防止同一个用户订阅两个分组，从而收到2个相同的消息。 continue; }
     * sendChannelHandlerContext.add(hashCode); //
     * ShareBean.getInstance().getChatRoomExecutorService() // .execute(new Runnable() { //
     * 
     * @Override // public void run() { try { Channel channel = channelHandlerContext.getChannel();
     * if (channel.isOpen() && channel.isConnected()) { if (logger.isDebugEnabled()) {
     * logger.debug("发送消息：" + channelHandlerContext.hashCode()); }
     * channel.write(textWebSocketFrame); } } catch (Exception e) { logger.error("发送消息异常" +
     * channelHandlerContext.hashCode(), e);
     * _clearChannelHandlerContexts.add(channelHandlerContext); } // } // }); } } if
     * (logger.isInfoEnabled()) { StringBuffer sf = new StringBuffer();
     * sf.append(this.chatRoomName).append("发送消息完毕 "); logger.info(sf.toString()); }
     * this.clearIt(_clearChannelHandlerContexts); }
     */

    public int getConnectionCount() {
        int size = 0;
        Set<String> ks = groupsChannelHandlerContext.keySet();
        for (String key : ks) {
            size += groupsChannelHandlerContext.get(key).size();
        }
        return size;
    }

    public String getChatRoomName() {
        return chatRoomName;
    }

    public void setChatRoomName(String chatRoomName) {
        this.chatRoomName = chatRoomName;
    }

}
